package com.atguigu.bigdata.spark.core.rdd.operator.transform;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.sources.In;
import org.datanucleus.store.rdbms.query.AbstractRDBMSQueryResult;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

public class Spark02_RDD_Operator_Transform_JAVA {
    public static void main(String[] args) {
        // TODO 准备环境
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("sparkCore");
        JavaSparkContext sc = new JavaSparkContext(conf);

        //
        // TODO 创建RDD
        // 从内存中创建RDD，将内存中集合的数据作为处理的数据源  使用parallelize
        List<Integer> seq = Arrays.asList(1,2,3,4,5,6);
        JavaRDD<Integer> rdd = sc.parallelize(seq,2);

        // mapPartitions : 可以以分区为单位进行数据转换操作
        //                 但是会将整个分区的数据加载到内存进行引用
        //                 如果处理完的数据是不会被释放掉，存在对象的引用。
        //                 在内存较小，数据量较大的场合下，容易出现内存溢出。
        JavaRDD<Integer> mapPartionRDD = rdd.mapPartitions(new FlatMapFunction<Iterator<Integer>, Integer>() {
            @Override
            public Iterator<Integer> call(Iterator<Integer> integerIterator) throws Exception {
                System.out.println("============>>>>>>>>>>>>>>>>>>>>>>");
                List<Integer> temp = new ArrayList<Integer>();
                while(integerIterator.hasNext()) {
                    Integer val = integerIterator.next();
                    temp.add(val * 2);
                }
                return temp.iterator();
            }
        });

        List<Integer> res = mapPartionRDD.collect();
        for(Integer val : res) {
            System.out.println(val);
        }

        sc.stop();

    }
}
